#include "RelOp.h"

void GroupBy::Run(Pipe &inPipe, Pipe &outPipe, OrderMaker &groupAtts, Function &computeMe){

    in = &inPipe;
    out = &outPipe;
    om = &groupAtts;
    func = &computeMe;

    pthread_create(&thread, NULL, GroupByWorker, (void*)this);
}

void GroupBy::WaitUntilDone(){
    pthread_join (thread, NULL);
}

void GroupBy::Use_n_Pages(int n){
    nOfPages = n;
}

void GroupBy::DoWork(){

     if (nOfPages == 0)
            nOfPages = 15;

    Pipe outSorted(100);
    BigQ q(*in, outSorted, *om, nOfPages);

    ComparisonEngine e;

    Record temp;
    Record lastMatch;
    int intResult;
    double doubleResult;
    double result = 0;
    Type t;

    bool b = outSorted.Remove(&temp);
    if (b){
        //om used to do comparison
        OrderMaker comparisonOM;
        comparisonOM.numAtts = om->numAtts;
        for (int i = 0; i < om->numAtts; i++){
            comparisonOM.whichAtts[i] = i;
            comparisonOM.whichTypes[i] = om->whichTypes[i];
        }

        //get the records info
        int numAttsLeft = 1;
        int numAttsRight = comparisonOM.numAtts;
        int numAttsToKeep = numAttsLeft + numAttsRight;
        int attsToKeep[numAttsToKeep];

        attsToKeep[0] = 0;
        for(int i = 1; i < comparisonOM.numAtts; i++)
                attsToKeep[i] = comparisonOM.whichAtts[i];

        int numAttsInput = ((((int*)(temp.bits))[1])/sizeof(int)) - 1;
        t = func->Apply(temp, intResult, doubleResult);

        temp.Project(om->whichAtts, om->numAtts, numAttsInput);
        lastMatch.Copy(&temp);

        if(t == Int)
            result += intResult;
        else
            result += doubleResult;

        while(outSorted.Remove(&temp)){
            t = func->Apply(temp, intResult, doubleResult);
            temp.Project(om->whichAtts, om->numAtts, numAttsInput);

            int comp = e.Compare(&lastMatch, &temp, &comparisonOM);

            if (comp == 0){
                if(t == Int)
                    result += intResult;
                else
                    result += doubleResult;
            }
            else{
                Attribute a = {"double", Double};
                Schema schema ("schema", 1, &a);

                std::stringstream s;
                s << result << "|";

                std::string str = s.str();
                char * writable = new char[str.size() + 1];
                std::copy(str.begin(), str.end(), writable);
                writable[str.size()] = '\0';

                Record sum;
                sum.ComposeRecord(&schema, writable);

                Record *joinedRecord = new Record;
                joinedRecord->MergeRecords(&sum, &lastMatch, numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, numAttsLeft);
                out->Insert(joinedRecord);

                result = 0;
                if(t == Int)
                    result += intResult;
                else
                    result += doubleResult;

                lastMatch.Copy(&temp);

                delete[] writable;

            }
        }

        Attribute a = {"double", Double};
        Schema schema ("schema", 1, &a);

        std::stringstream s;
        s << result << "|";

        std::string str = s.str();
        char * writable = new char[str.size() + 1];
        std::copy(str.begin(), str.end(), writable);
        writable[str.size()] = '\0';

        Record sum;
        sum.ComposeRecord(&schema, writable);

        Record *joinedRecord = new Record;
        joinedRecord->MergeRecords(&sum, &lastMatch, numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, numAttsLeft);
        out->Insert(joinedRecord);

        delete[] writable;
    }


    out->ShutDown();
}

void *GroupByWorker(void *param){
     GroupBy groupBy = *((GroupBy*) param);
     groupBy.DoWork();
}
